package cn.edu.shu;
  
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Answer{

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        env.setParallelism(1);
        // 创建一个输入表SourceTable
        String sourceDDL =
                "create table relation (" +
                        "referrer INT," +
                        "referree INT" +
                        ") with (" +
                        " 'connector' = 'filesystem', " +
                        " 'path' = 'file:///home/ubuntu/small_relation.csv', " +
                        " 'format' = 'csv', " +
                        " 'csv.field-delimiter' = ',', " +
                        " 'csv.ignore-parse-errors' = 'true' " +
                        " )";
        String dql="WITH website_sizes AS (SELECT referrer AS website, COUNT (DISTINCT referree) AS total_links FROM relation GROUP BY referrer) intersection AS (SELECT w1.referrer AS website_1, w2.referrer AS website_2, COUNT (DISTINCT w1.referree) AS intersection_size FROM relation w1 JOIN relation w2 ON w1.referree = w2.referree WHERE w1.referrer < w2.referrer GROUP BY w1.referrer, w2.referrer) select i.website_1, i.website_2, SELECT i.website_1, i.website_2, CASE WHEN ws1.total_links + ws2.total_links - i.intersection_size > 0 THEN i.intersection_size * 1.0 / (ws1.total_links + ws2.total_links - i.intersection_size) END AS similarity FROM intersection i JOIN website_sizes ws1 ON i.website_1 = ws1.website JOIN website_sizes ws2 ON i.website_2 = ws2.website"
        tableEnv.executeSql(sourceDDL);
        // 使用SQL语句创建一个输出表SinkTable
        tableEnv.executeSql("CREATE TEMPORARY TABLE SinkTable(stockId STRING,price DOUBLE) WITH ('connector' = 'print')");
        // 使用Table API创建一个Table对象table1
        // Table table1 = tableEnv.from("stockPriceTable");

        Table table2 = tableEnv.sqlQuery(dql);
        // 把Table对象table1写入到输出表SinkTable中
        TableResult tableResult = table2.executeInsert("SinkTable");
    }

}
